package org.ws.task.item;

import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.event.EventListener;
import org.ws.task.event.support.BootstrapEventHolder;
import org.ws.task.event.support.startup.BootstrapEvent;
import org.ws.task.event.support.startup.BootstrapEventType;
import org.ws.task.listener.item.TaskItemListener;
import org.ws.task.listener.TaskItemListenerManager;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.repository.TaskRepository;
import org.ws.task.model.protocol.CommitRequest;
import org.ws.task.model.protocol.CommitResponse;
import org.ws.task.model.TaskItem;
import org.ws.task.repository.WorkerContextRepository;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 抽象的Task
 * 实现公共功能
 */
public abstract class AbstractTask implements Task {

    private static Logger log = LoggerFactory.getLogger(AbstractTask.class.getName());

    private final String name;
    protected RemoteAccessApi remoteAccessApi;
    protected TaskInstance taskInstance;
    protected ExecuteContext context;
    private final TaskItemListenerManager taskItemListenerManager;
    private final CountDownLatch startCountDownLatch;

    public AbstractTask(String name) {
        this.name = name;
        this.taskItemListenerManager = new TaskItemListenerManager();
        BootstrapEventHolder.getInstance().registerListener(new EventListener<BootstrapEvent>(){
            @Override
            public void onEvent(BootstrapEvent event) {
                if(event.getEventType() == BootstrapEventType.BOOTSTRAP_SUCCESS){
                    AbstractTask.this.startCountDownLatch.countDown();
                }
            }
        });
        this.startCountDownLatch = new CountDownLatch(1);
    }

    @Override
    public void start(RemoteAccessApi remoteAccessApi) {
        this.remoteAccessApi = remoteAccessApi;
        this.remoteAccessApi.setListenerManager(this.taskItemListenerManager);
        this.taskInstance = TaskRepository.getInstance().getTaskInstance(this.name);
        this.context = WorkerContextRepository.getInstance().getContext(this.name);
        log.info("task[{}] start.",this.name);
    }

    public boolean awaitStart(long timeout) {
        return Uninterruptibles.awaitUninterruptibly(this.startCountDownLatch,timeout, TimeUnit.MILLISECONDS);
    }

    @Override
    public void shutdown() {
        this.context.clear();
        TaskRepository.getInstance().removeTaskInstance(this.name);
        this.context = null;
        this.taskInstance = null;
    }

    @Override
    public String name() {
        return name;
    }

    @Override
    public void registerListener(TaskItemListener listener) {
        this.taskItemListenerManager.registerListener(listener);
    }

    @Override
    public CommitResponse commitSync(List<TaskItem> items, long timeout) {
        CommitRequest request = buildCommitRequest(items);
        return this.remoteAccessApi.commitStatusSync(request, timeout);
    }

    @Override
    public void commitAsync(List<TaskItem> items, long timeout, Consumer<CommitResponse> callback) {
        CommitRequest request = buildCommitRequest(items);
        this.remoteAccessApi.commitStatusAsync(request,timeout,callback);
    }

    public RemoteAccessApi getRemoteAccessApi() {
        return remoteAccessApi;
    }

    public TaskItemListenerManager getTaskItemListenerManager() {
        return taskItemListenerManager;
    }

    private CommitRequest buildCommitRequest(List<TaskItem> items){
        String nodeId = NodeMetadataRepository.getInstance().getTheWorkerNode().getId();
        return new CommitRequest(this.name,items,nodeId);
    }

}
